# -*- coding: utf-8 -*-
from datetime import timedelta
from airflow.operators.bash import BashOperator
from jms_hi.dm.dm_jqr_cross_belt_sort_cnt_dt import jms_hi_dm__dm_jqr_cross_belt_sort_cnt_dt
from airflow.exceptions import AirflowSkipException
import pendulum

cst = pendulum.timezone('Asia/Shanghai')


class HiRichSqlSensor(BashOperator):
    def pre_execute(self, context):
        day = cst.convert(context['ti'].execution_date) + timedelta(days=1)

        days_of_hours = ['06']

        if day.strftime('%H') not in days_of_hours:
            print(f'{day.strftime("%H")} not in {days_of_hours}, should skip')
            raise AirflowSkipException()
        else:
            print(f'{day.strftime("%H")} in {days_of_hours}, run now')
            super().pre_execute(context)

jms_oracle__dm_jqr_cross_belt_sort_cnt_dt = HiRichSqlSensor(
    task_id='jms_oracle__dm_jqr_cross_belt_sort_cnt_dt',
    execution_timeout=timedelta(hours=1),
    email=['yl_etl@yl-scm.com','guoruiling@jtexpress.com'],
    # name='jms_oracle__dm_center_departure_prediction_{{ execution_date | date_add(1) | cst_ds }}',
    bash_command="jms_hi/oracle/dm_jqr_cross_belt_sort_cnt_dt/execute.sh",
    pool='spmi_apack',
    retries=1,
    pool_slots=1,
)

jms_oracle__dm_jqr_cross_belt_sort_cnt_dt << [jms_hi_dm__dm_jqr_cross_belt_sort_cnt_dt]